import { ServiceException } from '../common/excpetion/serviceException';
import * as socketIOSvr from 'socket.io';
import * as redisAdapter from 'socket.io-redis';
import * as redis from 'redis';
import { BaseMessageModel } from "../common/messageModel/baseMessageModel";
import { config } from '../config';
import { Guid } from '../common';
import * as _EventEmitter from 'events';
const EventEmitter: any = _EventEmitter;

const redisClient = redis.createClient(config.redisConfig.port, config.redisConfig.host);
redisClient.select(13);

export class MessageService extends EventEmitter {
    private io: socketIOSvr.Server;
    private timer;
    private nspKeyValues: { nspName: string; nspIO: socketIOSvr.Namespace }[] = [];
    async Start(server) {
        this.io = socketIOSvr(server, {
            path: `/`,
            serveClient: false,
            // below are engine.IO options
            pingInterval: 10000,
            pingTimeout: 5000,
            cookie: false
        });
        this.io.adapter(redisAdapter({ host: config.redisConfig.host, port: config.redisConfig.port }));

        let nspIo = this.io.of('/ldc');
        nspIo.on('connection', (socket) => {
            console.log(socket.id + ' Join');
            socket.on('disconnect', (reason) => {
                let args: LDCOnlineChangedEventArgs = {
                    isOnline: false,
                    id: (<any>socket).storeId
                };
                this.emit('ldcOnlineChanged', args);
            });
            socket.on('error', this.ErrorHandle);
            socket.on('bindStore', (ldcObj: LDCOnlineChangedEventArgs) => {
                ldcObj.isOnline = true;
                ldcObj.ipAddress = socket.conn.remoteAddress;
                if (ldcObj.id) {
                    this.emit('ldcOnlineChanged', ldcObj);
                    (<any>socket).storeId = ldcObj.id;
                }
            });
        });
        this.nspKeyValues.push({
            nspName: 'ldc',
            nspIO: nspIo
        });

        this.SetTimer();
    }

    Broadcast(data: BaseMessageModel, nsp?: string) {
        if (nsp) {
            let item = this.nspKeyValues.find(x => x.nspName == nsp);
            item.nspIO.emit(MessageEventName.SvrInner, JSON.stringify(data));
        }
        else {
            for (let item of this.nspKeyValues) {
                item.nspIO.emit(MessageEventName.SvrInner, JSON.stringify(data));
            }
        }
    }

    BroadcastMQ(data: BaseMessageModel, storeIds?: string[], nsp?: string) {
        let has = false;
        if (!storeIds) has = true;
        for (let item of this.nspKeyValues) {
            if (storeIds) {
                for (let socket in item.nspIO.sockets) {
                    if (storeIds.find(x => x == (<any>item.nspIO.sockets[socket]).storeId)) {
                        has = true;
                        break;
                    }
                }
            }

            if (has) {
                data.id = Guid.GetGuid();
                data.target = item.nspName;
                this.PushToQueue(MessageQueueType.Wait, data);
            }
        }

        this.SetTimer();
    }

    SendToStore(data, storeId?: string) {
        let timer;
        let timerHandle;
        return new Promise((resolve, reject) => {
            timerHandle = () => {
                reject({ message: '通讯超时！' });
            }
            let item = this.nspKeyValues.find(x => x.nspName == 'ldc');
            let hasStore = false;
            for (let key in item.nspIO.sockets) {
                let socket = item.nspIO.sockets[key];
                if ((<any>item.nspIO.sockets[key]).storeId == storeId) {
                    hasStore = true;
                    socket.emit(MessageEventName.RPCData, JSON.stringify(data), (res) => {
                        clearInterval(timer);
                        timer = null;
                        if (res.statusCode && res.statusCode >= 400 && res.statusCode <= 503) {
                            reject(res);
                        }
                        else {
                            resolve(res);
                        }
                    });
                    break;
                }
            }
            if (!hasStore) {
                throw new ServiceException('当前店铺未开通此服务！');
            }
            timer = setInterval(timerHandle, 10 * 1000);
        });
    }

    private async SendMessage() {
        this.CleanTimer();
        try {
            let keys = await this.GetRedisKeys(MessageQueueType.Wait);
            if (keys.length > 0) {
                for (let key of keys) {
                    let value = await this.GetRedisValue(key, MessageQueueType.Wait);
                    let msgData = JSON.parse(value) as BaseMessageModel;
                    try {
                        if (msgData.sendCount >= 5) {
                            redisClient.hdel(MessageQueueType.Wait, key);
                            this.PushToQueue(MessageQueueType.Error, msgData);
                        }
                        else {
                            let nsp = this.nspKeyValues.find(x => x.nspName == 'ldc');
                            if (nsp) {
                                msgData.sendCount++;
                                this.SetRedisValue(key, MessageQueueType.Wait, msgData);

                                for (let sKey in nsp.nspIO.sockets) {
                                    let socket = nsp.nspIO.sockets[sKey];
                                    socket.emit(MessageEventName.SvrInner, JSON.stringify(msgData), (args) => {
                                        redisClient.hdel(MessageQueueType.Wait, key);
                                    });
                                }
                            }
                        }
                    }
                    catch (error) {
                        console.error(error);
                        redisClient.hdel(MessageQueueType.Wait, key);
                        this.PushToQueue(MessageQueueType.Error, msgData);
                    }
                }
                this.SetTimer();
            }

        } catch (error) {
            console.error(error);
            this.SetTimer();
        }
    }

    private ErrorHandle(error) {
        //TODO:
        console.log('error', error);
    }

    private SetTimer() {
        if (!this.timer)
            this.timer = setInterval(this.SendMessage.bind(this), 1000);
    }

    private CleanTimer() {
        clearInterval(this.timer);
        this.timer = null;
    }

    private GetRedisKeys(key: string): Promise<string[]> {
        return new Promise((resolve, reject) => {
            redisClient.hkeys(key, (err, keys) => {
                if (err) {
                    reject(err);
                }
                else {
                    resolve(keys);
                }
            });
        });
    }

    private GetRedisValue(key: string, qKey: string): Promise<string> {
        return new Promise((resolve, reject) => {
            redisClient.hget(qKey, key, (err, value: string) => {
                if (err) reject(err);
                else {
                    resolve(value);
                }
            });
        })
    }

    private SetRedisValue(key: string, queueType: string, data) {
        redisClient.hset(queueType, key, JSON.stringify(data));
    }

    private PushToQueue(queueType: string, messageData: BaseMessageModel) {
        redisClient.hset(queueType, messageData.id, JSON.stringify(messageData));
    }
}

export const MessageEventName = {
    SvrInner: 'svrInner',
    RPCData: 'rpc.business'
}

export const MessageNamespace = {
    LDC: 'ldc'
}

const MessageQueueType = {
    Wait: 'message_wait',
    Error: 'message_error'
}

export interface LDCOnlineChangedEventArgs {
    isOnline: boolean;
    ipAddress?: string;
    id: string;
    ldc?: {
        version: string;
    }
};